package org.dromara.common.oss.core

import cn.hutool.core.io.FileUtil
import cn.hutool.core.io.IoUtil
import cn.hutool.core.util.IdUtil
import lombok.extern.slf4j.Slf4j
import org.dromara.common.core.constant.Constants
import org.dromara.common.core.utils.DateUtils.datePath
import org.dromara.common.core.utils.StringUtils
import org.dromara.common.oss.constant.OssConstant
import org.dromara.common.oss.entity.UploadResult
import org.dromara.common.oss.enums.AccessPolicyType
import org.dromara.common.oss.exception.OssException
import org.dromara.common.oss.properties.OssProperties
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider
import software.amazon.awssdk.core.async.AsyncResponseTransformer
import software.amazon.awssdk.core.async.BlockingInputStreamAsyncRequestBody
import software.amazon.awssdk.core.async.ResponsePublisher
import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient
import software.amazon.awssdk.regions.Region
import software.amazon.awssdk.services.s3.S3AsyncClient
import software.amazon.awssdk.services.s3.S3Configuration
import software.amazon.awssdk.services.s3.model.GetObjectResponse
import software.amazon.awssdk.services.s3.presigner.S3Presigner
import software.amazon.awssdk.transfer.s3.S3TransferManager
import software.amazon.awssdk.transfer.s3.model.Download
import software.amazon.awssdk.transfer.s3.model.DownloadRequest
import software.amazon.awssdk.transfer.s3.progress.LoggingTransferListener
import java.io.*
import java.net.URI
import java.nio.ByteBuffer
import java.nio.channels.Channels
import java.nio.file.Files
import java.nio.file.Path
import java.time.Duration
import java.util.*
import java.util.function.Consumer

/**
 * S3 存储协议 所有兼容S3协议的云厂商均支持
 * 阿里云 腾讯云 七牛云 minio
 *
 * @author AprilWind
 * @updater LikeYouDo
 * @date 2025/1/8 16:17
 */
@Slf4j
class OssClient private constructor() {

    /**
     * 服务商
     */
    private lateinit var configKey: String

    /**
     * 配置属性
     */
    private lateinit var properties: OssProperties

    /**
     * Amazon S3 异步客户端
     */
    private lateinit var client: S3AsyncClient

    /**
     * 用于管理 S3 数据传输的高级工具
     */
    private lateinit var transferManager: S3TransferManager

    /**
     * AWS S3 预签名 URL 的生成器
     */
    private lateinit var presigner: S3Presigner

    /**
     * 构造方法
     *
     * @param configKey     配置键
     * @param ossProperties Oss配置属性
     */
    constructor(configKey: String, ossProperties: OssProperties) : this() {
        this.configKey = configKey
        this.properties = ossProperties
        try {
            // 创建 AWS 认证信息
            val credentialsProvider = StaticCredentialsProvider.create(
                AwsBasicCredentials.create(properties.accessKey, properties.secretKey)
            )

            // MinIO 使用 HTTPS 限制使用域名访问，站点填域名。需要启用路径样式访问
            val isStyle = !StringUtils.containsAny(properties.endpoint, *OssConstant.CLOUD_SERVICE)

            // 创建AWS基于 Netty 的 S3 客户端
            this.client = S3AsyncClient.builder()
                .credentialsProvider(credentialsProvider)
                .endpointOverride(URI.create(getEndpoint()))
                .region(of())
                .forcePathStyle(isStyle)
                .httpClient(
                    NettyNioAsyncHttpClient.builder()
                        .connectionTimeout(Duration.ofSeconds(60)) // 设置连接超时
                        .build()
                )
                .build()

            // AWS基于 CRT 的 S3 AsyncClient 实例用作 S3 传输管理器的底层客户端
            this.transferManager = S3TransferManager.builder().s3Client(this.client).build()

            // 创建 S3 配置对象
            val config = S3Configuration.builder().chunkedEncodingEnabled(false)
                .pathStyleAccessEnabled(isStyle).build()

            // 创建 预签名 URL 的生成器 实例，用于生成 S3 预签名 URL
            this.presigner = S3Presigner.builder()
                .region(of())
                .credentialsProvider(credentialsProvider)
                .endpointOverride(URI.create(getDomain()))
                .serviceConfiguration(config)
                .build()
        } catch (e: Exception) {
            if (e is OssException) {
                throw e
            }
            throw OssException("配置错误! 请检查系统配置:[${e.message}]")
        }
    }

    /**
     * 上传文件到 Amazon S3，并返回上传结果
     *
     * @param filePath    本地文件路径
     * @param key         在 Amazon S3 中的对象键
     * @param md5Digest   本地文件的 MD5 哈希值（可选）
     * @param contentType 文件内容类型
     * @return UploadResult 包含上传后的文件信息
     * @throws OssException 如果上传失败，抛出自定义异常
     */
    fun upload(filePath: Path?, key: String, md5Digest: String?, contentType: String?): UploadResult {
        try {
            // 构建上传请求对象
            val fileUpload = transferManager.uploadFile { x ->
                x.putObjectRequest { y ->
                    y.bucket(properties.bucketName)
                        .key(key)
                        .contentMD5(if (StringUtils.isNotEmpty(md5Digest)) md5Digest else null)
                        .contentType(contentType)
                        // 用于设置对象的访问控制列表（ACL）。不同云厂商对ACL的支持和实现方式有所不同，
                        // 因此根据具体的云服务提供商，你可能需要进行不同的配置（自行开启，阿里云有acl权限配置，腾讯云没有acl权限配置）
                        //.acl(getAccessPolicy().getObjectCannedACL())
                        .build()
                }
                    .addTransferListener(LoggingTransferListener.create())
                    .source(filePath).build()
            }

            // 等待上传完成并获取上传结果
            val uploadResult = fileUpload.completionFuture().join()
            val eTag = uploadResult.response().eTag()

            // 提取上传结果中的 ETag，并构建一个自定义的 UploadResult 对象
            return UploadResult().apply {
                url = getUrl() + StringUtils.SLASH + key
                filename = key
                this.eTag = eTag
            }
        } catch (e: Exception) {
            // 捕获异常并抛出自定义异常
            throw OssException("上传文件失败，请检查配置信息:[${e.message}]")
        } finally {
            // 无论上传是否成功，最终都会删除临时文件
            FileUtil.del(filePath)
        }
    }

    /**
     * 上传 InputStream 到 Amazon S3
     *
     * @param inputStream 要上传的输入流
     * @param key         在 Amazon S3 中的对象键
     * @param length      输入流的长度
     * @param contentType 文件内容类型
     * @return UploadResult 包含上传后的文件信息
     * @throws OssException 如果上传失败，抛出自定义异常
     */
    fun upload(ist: InputStream?, key: String, length: Long?, contentType: String?): UploadResult {
        // 如果输入流不是 ByteArrayInputStream，则将其读取为字节数组再创建 ByteArrayInputStream
        var inputStream = ist
        if (inputStream !is ByteArrayInputStream) {
            inputStream = ByteArrayInputStream(IoUtil.readBytes(inputStream))
        }
        try {
            // 创建异步请求体（length如果为空会报错）
            val body = BlockingInputStreamAsyncRequestBody.builder()
                .contentLength(length)
                .subscribeTimeout(Duration.ofSeconds(120))
                .build()

            // 使用 transferManager 进行上传
            val upload = transferManager.upload { x ->
                x.requestBody(body)
                    .addTransferListener(LoggingTransferListener.create())
                    .putObjectRequest { y ->
                        y.bucket(properties.bucketName)
                            .key(key)
                            .contentType(contentType)
                            // 用于设置对象的访问控制列表（ACL）。不同云厂商对ACL的支持和实现方式有所不同，
                            // 因此根据具体的云服务提供商，你可能需要进行不同的配置（自行开启，阿里云有acl权限配置，腾讯云没有acl权限配置）
                            //.acl(getAccessPolicy().getObjectCannedACL())
                            .build()
                    }
                    .build()
            }

            // 将输入流写入请求体
            body.writeInputStream(inputStream)

            // 等待文件上传操作完成
            val uploadResult = upload.completionFuture().join()
            val eTag = uploadResult.response().eTag()

            // 提取上传结果中的 ETag，并构建一个自定义的 UploadResult 对象
            return UploadResult().apply {
                url = getUrl() + StringUtils.SLASH + key
                filename = key
                this.eTag = eTag
            }
        } catch (e: Exception) {
            throw OssException("上传文件失败，请检查配置信息:[${e.message}]")
        }
    }

    /**
     * 下载文件从 Amazon S3 到临时目录
     *
     * @param path 文件在 Amazon S3 中的对象键
     * @return 下载后的文件在本地的临时路径
     * @throws OssException 如果下载失败，抛出自定义异常
     */
    fun fileDownload(path: String): Path {
        // 构建临时文件
        val tempFilePath = FileUtil.createTempFile().toPath()
        // 使用 S3TransferManager 下载文件
        val downloadFile = transferManager.downloadFile { x ->
            x.getObjectRequest { y ->
                y.bucket(properties.bucketName)
                    .key(removeBaseUrl(path))
                    .build()
            }
                .addTransferListener(LoggingTransferListener.create())
                .destination(tempFilePath)
                .build()
        }
        // 等待文件下载操作完成
        downloadFile.completionFuture().join()
        return tempFilePath
    }

    /**
     * 下载文件从 Amazon S3 到 输出流
     *
     * @param key 文件在 Amazon S3 中的对象键
     * @param out 输出流
     * @param consumer 自定义处理逻辑
     * @throws OssException 如果下载失败，抛出自定义异常
     */
    fun download(key: String, out: OutputStream, consumer: Consumer<Long>) {
        try {
            this.download(key, consumer).writeTo(out)
        } catch (e: Exception) {
            throw OssException("文件下载失败，错误信息:[${e.message}]")
        }
    }

    /**
     * 下载文件从 Amazon S3 到 输出流
     *
     * @param key 文件在 Amazon S3 中的对象键
     * @param contentLengthConsumer 文件大小消费者函数
     * @return 写出订阅器
     * @throws OssException 如果下载失败，抛出自定义异常
     */
    fun download(key: String?, contentLengthConsumer: Consumer<Long>?): WriteOutSubscriber<OutputStream> {
        try {
            // 构建下载请求
            val publisherDownloadRequest: DownloadRequest<ResponsePublisher<GetObjectResponse>> = DownloadRequest.builder()
                // 文件对象
                .getObjectRequest { y ->
                    y.bucket(properties.bucketName)
                        .key(key)
                        .build()
                }
                .addTransferListener(LoggingTransferListener.create())
                // 使用发布订阅转换器
                .responseTransformer(AsyncResponseTransformer.toPublisher())
                .build()

            // 使用 S3TransferManager 下载文件
            val publisherDownload: Download<ResponsePublisher<GetObjectResponse>> = transferManager.download(publisherDownloadRequest)
            // 获取下载发布订阅转换器
            val publisher: ResponsePublisher<GetObjectResponse> = publisherDownload.completionFuture().join().result()
            // 执行文件大小消费者函数
            Optional.ofNullable(contentLengthConsumer)
                .ifPresent { lengthConsumer ->
                    lengthConsumer.accept(publisher.response().contentLength())
                }

            // 构建写出订阅器对象
            return WriteOutSubscriber { out: OutputStream ->
                Channels.newChannel(out).use { channel ->
                    // 订阅数据
                    publisher.subscribe(Consumer { byteBuffer: ByteBuffer ->
                        while (byteBuffer.hasRemaining()) {
                            try {
                                channel.write(byteBuffer)
                            } catch (e: IOException) {
                                throw RuntimeException(e)
                            }
                        }
                    }).join()
                }
            }
        } catch (e: java.lang.Exception) {
            throw OssException("文件下载失败，错误信息:[" + e.message + "]")
        }
    }

    /**
     * 删除云存储服务中指定路径下文件
     *
     * @param path 指定路径
     */
    fun delete(path: String) {
        try {
            client.deleteObject { x ->
                x.bucket(properties.bucketName)
                    .key(removeBaseUrl(path))
                    .build()
            }
        } catch (e: java.lang.Exception) {
            throw OssException("删除文件失败，请检查配置信息:[${e.message}]")
        }
    }

    /**
     * 获取私有URL链接
     *
     * @param objectKey 对象KEY
     * @param expiredTime    链接授权到期时间
     */
    fun getPrivateUrl(objectKey: String?, expiredTime: Duration): String {
        // 使用 AWS S3 预签名 URL 的生成器 获取对象的预签名 URL
        val url = presigner.presignGetObject { x ->
            x.signatureDuration(expiredTime)
                .getObjectRequest { y ->
                    y.bucket(properties.bucketName)
                        .key(objectKey)
                        .build()
                }
                .build()
        }
            .url()
        return url.toString()
    }

    /**
     * 上传 byte[] 数据到 Amazon S3，使用指定的后缀构造对象键。
     *
     * @param data   要上传的 byte[] 数据
     * @param suffix 对象键的后缀
     * @return UploadResult 包含上传后的文件信息
     * @throws OssException 如果上传失败，抛出自定义异常
     */
    fun uploadSuffix(data: ByteArray, suffix: String, contentType: String?): UploadResult =
        upload(ByteArrayInputStream(data), getPath(properties.prefix, suffix), data.size.toLong(), contentType)

    /**
     * 上传 InputStream 到 Amazon S3，使用指定的后缀构造对象键。
     *
     * @param inputStream 要上传的输入流
     * @param suffix      对象键的后缀
     * @param length      输入流的长度
     * @return UploadResult 包含上传后的文件信息
     * @throws OssException 如果上传失败，抛出自定义异常
     */
    fun uploadSuffix(inputStream: InputStream?, suffix: String, length: Long?, contentType: String?): UploadResult =
        upload(inputStream, getPath(properties.prefix, suffix), length, contentType)

    /**
     * 上传文件到 Amazon S3，使用指定的后缀构造对象键
     *
     * @param file   要上传的文件
     * @param suffix 对象键的后缀
     * @return UploadResult 包含上传后的文件信息
     * @throws OssException 如果上传失败，抛出自定义异常
     */
    fun uploadSuffix(file: File, suffix: String): UploadResult =
        upload(file.toPath(), getPath(properties.prefix, suffix), null, FileUtil.getMimeType(suffix))

    /**
     * 获取文件输入流
     *
     * @param path 完整文件路径
     * @return 输入流
     */
    @Throws(IOException::class)
    fun getObjectContent(path: String): InputStream {
        // 下载文件到临时目录
        val tempFilePath = fileDownload(path)
        // 创建输入流
        val inputStream = Files.newInputStream(tempFilePath)
        // 删除临时文件
        FileUtil.del(tempFilePath)
        // 返回对象内容的输入流
        return inputStream
    }

    /**
     * 获取 S3 客户端的终端点 URL
     *
     * @return 终端点 URL
     */
    fun getEndpoint(): String {
        // 根据配置文件中的是否使用 HTTPS，设置协议头部
        val header = getIsHttps()
        // 拼接协议头部和终端点，得到完整的终端点 URL
        return header + properties.endpoint
    }

    /**
     * 获取 S3 客户端的终端点 URL（自定义域名）
     *
     * @return 终端点 URL
     */
    fun getDomain(): String {
        // 从配置中获取域名、终端点、是否使用 HTTPS 等信息
        val domain = properties.domain
        val endpoint = properties.endpoint
        val header = getIsHttps()

        // 如果是云服务商，直接返回域名或终端点
        if (StringUtils.containsAny(endpoint, *OssConstant.CLOUD_SERVICE)) {
            return if (StringUtils.isNotEmpty(domain)) header + domain else header + endpoint
        }

        // 如果是 MinIO，处理域名并返回
        if (StringUtils.isNotEmpty(domain)) {
            return if (domain.startsWith(Constants.HTTPS) || domain.startsWith(Constants.HTTP)) domain else header + domain
        }

        // 返回终端点
        return header + endpoint
    }

    /**
     * 根据传入的 region 参数返回相应的 AWS 区域
     * 如果 region 参数非空，使用 Region.of 方法创建并返回对应的 AWS 区域对象
     * 如果 region 参数为空，返回一个默认的 AWS 区域（例如，us-east-1），作为广泛支持的区域
     *
     * @return 对应的 AWS 区域对象，或者默认的广泛支持的区域（us-east-1）
     */
    fun of(): Region {
        // AWS 区域字符串
        val region = properties.region
        // 如果 region 参数非空，使用 Region.of 方法创建对应的 AWS 区域对象，否则返回默认区域
        return if (StringUtils.isNotEmpty(region)) Region.of(region) else Region.US_EAST_1
    }

    /**
     * 获取云存储服务的URL
     *
     * @return 文件路径
     */
    fun getUrl(): String {
        val domain = properties.domain
        val endpoint = properties.endpoint
        val header = getIsHttps()
        // 云服务商直接返回
        if (StringUtils.containsAny(endpoint, *OssConstant.CLOUD_SERVICE)) {
            return header + (if (StringUtils.isNotEmpty(domain)) domain else properties.bucketName + "." + endpoint)
        }
        // MinIO 单独处理
        if (StringUtils.isNotEmpty(domain)) {
            // 如果 domain 以 "https://" 或 "http://" 开头
            return if ((domain.startsWith(Constants.HTTPS) || domain.startsWith(Constants.HTTP)))
                domain + StringUtils.SLASH + properties.bucketName
            else header + domain + StringUtils.SLASH + properties.bucketName
        }
        return header + endpoint + StringUtils.SLASH + properties.bucketName
    }

    /**
     * 生成一个符合特定规则的、唯一的文件路径。通过使用日期、UUID、前缀和后缀等元素的组合，确保了文件路径的独一无二性
     *
     * @param prefix 前缀
     * @param suffix 后缀
     * @return 文件路径
     */
    fun getPath(prefix: String, suffix: String): String {
        // 生成uuid
        val uuid = IdUtil.fastSimpleUUID()
        // 生成日期路径
        val datePath = datePath()
        // 拼接路径
        val path =
            if (StringUtils.isNotEmpty(prefix))
                prefix + StringUtils.SLASH + datePath + StringUtils.SLASH + uuid
            else datePath + StringUtils.SLASH + uuid
        return path + suffix
    }

    /**
     * 移除路径中的基础URL部分，得到相对路径
     *
     * @param path 完整的路径，包括基础URL和相对路径
     * @return 去除基础URL后的相对路径
     */
    fun removeBaseUrl(path: String): String =
        path.replace(getUrl() + StringUtils.SLASH, "")

    /**
     * 服务商
     */
    fun getConfigKey(): String = configKey

    /**
     * 获取是否使用 HTTPS 的配置，并返回相应的协议头部。
     *
     * @return 协议头部，根据是否使用 HTTPS 返回 "https://" 或 "http://"
     */
    fun getIsHttps(): String {
        return if (OssConstant.IS_HTTPS == properties.isHttps)
            Constants.HTTPS
        else Constants.HTTP
    }

    /**
     * 检查配置是否相同
     */
    fun checkPropertiesSame(properties: OssProperties): Boolean = this.properties == properties

    /**
     * 获取当前桶权限类型
     *
     * @return 当前桶权限类型code
     */
    fun getAccessPolicy(): AccessPolicyType = AccessPolicyType.getByType(properties.accessPolicy)

}
